package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo3Map {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setAppName("map")
      .setMaster("local")

    val sc = new SparkContext(conf)


    /**
      * 构建rdd的方法
      * 1、读取文件
      * 2、基于scala的集合构建rdd --- 用于测试
      *
      */


    //基于集合构建rdd
    val listRDD: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 2)

    println(listRDD.getNumPartitions)

    /**
      * map算子:
      * 将rdd中的数据一条一条传递给后面的函数。将函数的返回值构建成一个新的rdd
      * map不会产生shuffle,map之后rdd的分区数等于map之前rdd的分区数
      *
      *
      * map是一个转换算子-- 懒执行， 需要一个action触发执行
      * foreach： action算子
      *
      * 如果一个算子的返回值是一个新的rdd,那么这个算子就是转换算子
      */
    val mapRDD: RDD[Int] = listRDD.map(i => {
      println("map")
      i * 2
    })


    //mapRDD.foreach(println)

    /**
      * mapPartitions ： 将每一个分区的数据传递给后面的函数，函数需要返回一个迭代器，再构建一个新的rdd
      *
      * mapPartitions： 懒执行，转换算子
      */

    val mapPartitionsRDD: RDD[Int] = listRDD.mapPartitions((iter: Iterator[Int]) => {

      val iterator: Iterator[Int] = iter.map(i => i * 2)

      //最后一行作为返回值
      iterator
    }
    )

    //mapPartitionsRDD.foreach(println)


    /**
      * mapPartitionsWithIndex: 一次遍历一个分区，会多一个分区的下标
      *
      */

    val mapPartitionsWithIndexRDD: RDD[Int] = listRDD.mapPartitionsWithIndex((index: Int, iter: Iterator[Int]) => {
      println(s"当前分区的编号:$index")
      iter
    })

    mapPartitionsWithIndexRDD.foreach(println)


  }

}
